package kv

import (
	"database/sql"
	"fmt"
	"gitee.com/tziye/renai-go/common"
	"gitee.com/tziye/renai-go/sqlproxy"
	"time"
)

const (
	Pair      = "Pair"
	List      = "List"
	PriorityQ = "PriorityQ"
	Map       = "Map"
)

func TypeInfo(kvType string) [4]string {
	switch kvType {
	case Pair:
		return [4]string{"renai_kv_pair", "renai_kv_pair_valid", "", ""}
	case List:
		return [4]string{"renai_kv_list_key", "renai_kv_list_key_valid", "renai_kv_list_value", "renai_kv_list_value_valid"}
	case PriorityQ:
		return [4]string{"renai_kv_queue_key", "renai_kv_queue_key_valid", "renai_kv_queue_value", "renai_kv_queue_value_valid"}
	case Map:
		return [4]string{"renai_kv_map_key", "renai_kv_map_key_valid", "renai_kv_map_value", "renai_kv_map_value_valid"}
	default:
		return [4]string{}
	}
}

type RenaiKey struct {
	Id         int64
	K          string
	V          string
	ExpireTime time.Time
	CreateTime time.Time
	UpdateTime time.Time
}

type RenaiValue struct {
	Id         int64
	K          string
	SubK       string
	V          string
	Score      int64
	CreateTime time.Time
	UpdateTime time.Time
}

func saveKey(db *sql.DB, key string, expireSeconds int, kvType string) {
	typeInfo := TypeInfo(kvType)
	if expireSeconds <= 0 {
		s := `INSERT INTO %s(k) VALUES (?) ON DUPLICATE KEY UPDATE 
              	expire_time = IF(expire_time <= NOW(), NULL, expire_time), update_time = NOW()`
		s = fmt.Sprintf(s, typeInfo[0])
		sqlproxy.Exec(db, s, key)
	} else {
		s := `INSERT INTO %s(k, expire_time) VALUES (?, DATE_ADD(NOW(), INTERVAL ? SECOND)) 
              	ON DUPLICATE KEY UPDATE expire_time = VALUES(expire_time), update_time = NOW()`
		s = fmt.Sprintf(s, typeInfo[0])
		sqlproxy.Exec(db, s, key, expireSeconds)
	}
}

func queryKey(db *sql.DB, key string, kvType string) RenaiKey {
	typeInfo := TypeInfo(kvType)
	s := "SELECT * FROM %s WHERE k = ?"
	s = fmt.Sprintf(s, typeInfo[1])
	var renaiKey RenaiKey
	row := sqlproxy.QueryRow(db, s, key)
	var expireTime sql.NullTime
	err := row.Scan(&renaiKey.Id, &renaiKey.K, &renaiKey.V, &expireTime, &renaiKey.CreateTime, &renaiKey.UpdateTime)
	if expireTime.Valid {
		renaiKey.ExpireTime = expireTime.Time
	}
	common.LogErr(err)
	return renaiKey
}

func queryKeys(db *sql.DB, keys []string, kvType string) []RenaiKey {
	typeInfo := TypeInfo(kvType)
	if len(keys) == 0 {
		return []RenaiKey{}
	}
	s := "SELECT * FROM %s WHERE k IN ("
	params := make([]any, 0)
	for i := 0; i < len(keys)-1; i++ {
		s += "?,"
		params = append(params, keys[i])
	}
	s += "?)"
	params = append(params, keys[len(keys)-1])
	s = fmt.Sprintf(s, typeInfo[1])
	rows := sqlproxy.QueryRows(db, s, params...)
	return fetchKey(rows)
}

func keysByType(db *sql.DB, kvType string) []RenaiKey {
	typeInfo := TypeInfo(kvType)
	s := "SELECT * FROM %s"
	s = fmt.Sprintf(s, typeInfo[1])
	rows := sqlproxy.QueryRows(db, s)
	return fetchKey(rows)
}

func keysLike(db *sql.DB, pattern string, kvType string) []RenaiKey {
	typeInfo := TypeInfo(kvType)
	s := "SELECT * FROM %s WHERE k LIKE ?"
	s = fmt.Sprintf(s, typeInfo[1])
	rows := sqlproxy.QueryRows(db, s, pattern)
	return fetchKey(rows)
}

func keysRegexp(db *sql.DB, pattern string, kvType string) []RenaiKey {
	typeInfo := TypeInfo(kvType)
	s := "SELECT * FROM %s WHERE k REGEXP ?"
	s = fmt.Sprintf(s, typeInfo[1])
	rows := sqlproxy.QueryRows(db, s, pattern)
	return fetchKey(rows)
}

func fetchKey(rows *sql.Rows) []RenaiKey {
	defer rows.Close()
	keys := make([]RenaiKey, 0)
	for rows.Next() {
		var key RenaiKey
		var expireTime sql.NullTime
		err := rows.Scan(&key.Id, &key.K, &key.V, &expireTime, &key.CreateTime, &key.UpdateTime)
		if expireTime.Valid {
			key.ExpireTime = expireTime.Time
		}
		common.PanicErr(err)
		keys = append(keys, key)
	}
	return keys
}

func values(db *sql.DB, key string, kvType string) []RenaiValue {
	typeInfo := TypeInfo(kvType)
	s := "SELECT * FROM %s WHERE k = ? ORDER BY %s"
	orderBy := "score"
	if kvType == Map {
		orderBy = "update_time"
	}
	s = fmt.Sprintf(s, typeInfo[3], orderBy)
	rows := sqlproxy.QueryRows(db, s, key)
	return fetchValue(rows, kvType == Map)
}

func fetchValue(rows *sql.Rows, isMap bool) []RenaiValue {
	defer rows.Close()
	values := make([]RenaiValue, 0)
	for rows.Next() {
		var value RenaiValue
		if isMap {
			err := rows.Scan(&value.Id, &value.K, &value.SubK, &value.V, &value.CreateTime, &value.UpdateTime)
			common.PanicErr(err)
		} else {
			err := rows.Scan(&value.Id, &value.K, &value.V, &value.Score, &value.CreateTime, &value.UpdateTime)
			common.PanicErr(err)
		}
		values = append(values, value)
	}
	return values
}

func total(db *sql.DB, kvType string) int {
	typeInfo := TypeInfo(kvType)
	s := "SELECT COUNT(id) FROM %s"
	s = fmt.Sprintf(s, typeInfo[1])
	var count sql.NullInt32
	row := sqlproxy.QueryRow(db, s)
	err := row.Scan(&count)
	common.PanicErr(err)
	return int(count.Int32)
}

func size(db *sql.DB, key string, kvType string) int {
	typeInfo := TypeInfo(kvType)
	s := "SELECT COUNT(id) FROM %s WHERE k = ?"
	s = fmt.Sprintf(s, typeInfo[3])
	var count sql.NullInt32
	row := sqlproxy.QueryRow(db, s, key)
	err := row.Scan(&count)
	common.PanicErr(err)
	return int(count.Int32)
}

func expire(db *sql.DB, key string, expireSeconds int, kvType string) {
	typeInfo := TypeInfo(kvType)
	s := `INSERT INTO %s(k, expire_time) VALUES (?, DATE_ADD(NOW(), INTERVAL ? SECOND)) 
        	ON DUPLICATE KEY UPDATE expire_time = VALUES(expire_time), update_time = NOW()`
	s = fmt.Sprintf(s, typeInfo[0])
	sqlproxy.Exec(db, s, key, expireSeconds)
}

func exists(db *sql.DB, key string, kvType string) bool {
	typeInfo := TypeInfo(kvType)
	s := "SELECT COUNT(id) FROM %s WHERE k = ?"
	s = fmt.Sprintf(s, typeInfo[1])
	var count sql.NullInt32
	row := sqlproxy.QueryRow(db, s, key)
	err := row.Scan(&count)
	common.PanicErr(err)
	return count.Int32 > 0
}

func contains(db *sql.DB, key string, value any, kvType string) bool {
	typeInfo := TypeInfo(kvType)
	s := "SELECT COUNT(id) FROM %s WHERE k = ? AND v = ?"
	s = fmt.Sprintf(s, typeInfo[3])
	var count sql.NullInt32
	row := sqlproxy.QueryRow(db, s, key, common.String(value))
	err := row.Scan(&count)
	common.PanicErr(err)
	return count.Int32 > 0
}

func clear(db *sql.DB, key string, kvType string) int {
	typeInfo := TypeInfo(kvType)
	s := "DELETE FROM %s WHERE k = ?"
	s = fmt.Sprintf(s, typeInfo[2])
	return sqlproxy.Exec(db, s, key)
}

func remove(db *sql.DB, key string, value any, kvType string) bool {
	typeInfo := TypeInfo(kvType)
	s := "DELETE FROM %s WHERE k = ? AND v = ?"
	s = fmt.Sprintf(s, typeInfo[2])
	return sqlproxy.Exec(db, s, key, common.String(value)) > 0
}

func batchRemove(db *sql.DB, key string, values []any, kvType string) int {
	typeInfo := TypeInfo(kvType)
	if len(values) == 0 {
		return 0
	}
	s := "DELETE FROM %s WHERE k = ? AND v IN ("
	params := []any{key}
	for i := 0; i < len(values)-1; i++ {
		s += "?,"
		params = append(params, common.String(values[i]))
	}
	s += "?)"
	params = append(params, common.String(values[len(values)-1]))
	s = fmt.Sprintf(s, typeInfo[2])
	return sqlproxy.Exec(db, s, params...)
}

func delete(db *sql.DB, key string, kvType string) {
	typeInfo := TypeInfo(kvType)
	s := "DELETE FROM %s WHERE k = '%s'"
	s1 := fmt.Sprintf(s, typeInfo[0], key)
	sqlList := []string{s1}
	if typeInfo[2] != "" {
		s2 := fmt.Sprintf(s, typeInfo[2], key)
		sqlList = append(sqlList, s2)
	}
	sqlproxy.BatchExec(db, sqlList)
}

func batchDelete(db *sql.DB, keys []string, kvType string) {
	typeInfo := TypeInfo(kvType)
	if len(keys) == 0 {
		return
	}
	s := "DELETE FROM %s WHERE k IN ("
	params := []any{typeInfo[0]}
	for i := 0; i < len(keys)-1; i++ {
		s += "'%s',"
		params = append(params, common.String(keys[i]))
	}
	s += "'%s')"
	params = append(params, common.String(keys[len(keys)-1]))
	s1 := fmt.Sprintf(s, params...)
	sqlList := []string{s1}
	if typeInfo[2] != "" {
		params[0] = typeInfo[2]
		s2 := fmt.Sprintf(s, params...)
		sqlList = append(sqlList, s2)
	}
	sqlproxy.BatchExec(db, sqlList)
}

func flush(db *sql.DB, kvType string) {
	typeInfo := TypeInfo(kvType)
	s := "DELETE FROM %s WHERE 1=1"
	s1 := fmt.Sprintf(s, typeInfo[0])
	sqlList := []string{s1}
	if typeInfo[2] != "" {
		s2 := fmt.Sprintf(s, typeInfo[2])
		sqlList = append(sqlList, s2)
	}
	sqlproxy.BatchExec(db, sqlList)
}
